// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include "coin2/exchange/huobi/feed_v1/processor.h"

#include <string_view>

#include <rapidjson/error/en.h>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/huobi/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::huobi::feed_v1::impl {
void HuobiFeedProcessor::OnTopicRecord(const TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error("JSON parse error: "s + GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.HasMember("ch")) {
    std::string ch = document["ch"].GetString();
    auto channel = HuobiChannel::Parse(ch);

    // pass kline channel
    // kline postfix is interval
    if (channel.type.rfind("kline.") == 0 && document.HasMember("data")) {
      // kline rest channel, response channel is kline.1day
      // change to period=1day
      channel.type = channel.type.replace(0, 6, "period=");
      if (!CheckChannelFilter("symbol={}&" + channel.type)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }
    } else {
      // parse this msg?
      // e.g. market.{}.trade.detail
      if (!CheckChannelFilter("market.{}." + channel.type)) {
        // WEMIX mbp.150 case is special.
        if (channel.native_symbol == "wemixusdt" &&
            channel.type == "mbp.150") {
          if (!CheckChannelFilter("market.{}.mbp.5")) {
            upd_mgr_.onEndPacket(packet);
            return;
          }
        } else {
          upd_mgr_.onEndPacket(packet);
          return;
        }
      }
    }

    std::string_view native_symbol = channel.native_symbol;
    if (channel.type == "trade.detail") {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }
      // trades:
      //  symbol, ts
      //   p, q, side
      //   p, q, side
      //   p, q, side
      // ...

      if (parser_config_.has_trade_sampling_frequency_bps()) {
        auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
        int64_t timestamp = stoull(document["ts"].GetString()) * 1'000'000LL;  // ms -> ns
        for (const auto& json_trade : document["tick"]["data"].GetArray()) {
          Trade _trade{};
          _trade.timestamp = timestamp;
          HuobiFeedUpdateParser::ParseTrade(json_trade, 1, &_trade);
          bool aggregated = slim_trade->TryAggregateSlimTrade(
                  packet.timestamp,
                  _trade.price,
                  _trade.fill_qty,
                  _trade.side,
                  _trade.timestamp,
                  parser_config_.trade_sampling_frequency_bps());
          int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
          if (!aggregated ||
              timespan > parser_config_.trade_sampling_time_length()) {
            auto* trade = upd_mgr_.GetTradeMutable();
            CHECK(slim_trade->PopulateTradeAndClear(trade));
            upd_mgr_.PublishTrade(native_symbol, false);
          }
        } 
      } else {
        Trade* agg_trade = upd_mgr_.GetTradeMutable();
        agg_trade->Clear();
        Trade trade{};

        agg_trade->timestamp = stoull(document["ts"].GetString()) * 1'000'000LL;  // ms -> ns
        // Huobi trade timestamp is reused, not updated by every trades array.
        trade.timestamp = agg_trade->timestamp;
        const rapidjson::Value& trades = document["tick"]["data"];
        for (const auto& trade_json : trades.GetArray()) {
          HuobiFeedUpdateParser::ParseTrade(trade_json, 1, &trade);
          if (TryAggregateTrades(agg_trade, trade)) {
            continue;
          }
          if (agg_trade->fill_qty > 0) {
            upd_mgr_.PublishTrade(native_symbol, true);
          }
          *agg_trade = trade;
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTrade(native_symbol, false);
        }
      }
    } else if (channel.type == "depth.step0") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        HuobiFeedUpdateParser::ParseBook(book, document, true);
        upd_mgr_.PublishBook();
      }
    } else if (channel.type.find(R"(mbp.)") == 0) {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        HuobiFeedUpdateParser::ParseBook(book, document, false);
        upd_mgr_.PublishBook();
      }
    } else if (channel.type == "bbo") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        HuobiFeedUpdateParser::ParseBbo(book, document);
        if (parser_config_.has_sampling_frequency_bps()) {
          bool sampled = book->SampleByFrequency(
                  book->Bid0(),
                  book->Ask0(),
                  parser_config_.sampling_frequency_bps());
          if (parser_config_.has_trade_sampling_frequency_bps()) {
            auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
            int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
            if (sampled ||
                timespan > parser_config_.trade_sampling_time_length()) {
              auto* trade = upd_mgr_.GetTradeMutable();
              if (slim_trade->PopulateTradeAndClear(trade)) {
                upd_mgr_.PublishTrade(native_symbol, false);
              }
            }
          }
          if (sampled) {
            upd_mgr_.PublishBook();
          }
        } else {
          upd_mgr_.PublishBook();
        }
      }
    } else if (channel.type == "etp") {
      if (upd_mgr_.is_symbol_registered(native_symbol)) {
        auto* nav = upd_mgr_.GetNavMutable();
        HuobiFeedUpdateParser::ParseNav(document, nav);
        upd_mgr_.PublishNav(native_symbol);
      }
    } else if (channel.type.rfind("period=") == 0) {
      // kline rest
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }

      // market.BTC-USD.kline.1day
      std::string_view period = ch.substr(ch.find_last_of('.') + 1);
      auto interval = HuobiFeedUpdateParser::ParseKlineInterval(period);
      auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (klines == nullptr) return;

      int64_t ts = stoull(document["ts"].GetString()) * 1'000'000LL;

      bool to_publish = false;
      if (document.HasMember("data")) {
        const auto& data = document["data"].GetArray();
        // from rest api
        if (data.Size() > 10) {
          klines->SnapshotClear();
        } else {
          klines->UpdateClear();
        }
        for (int i = data.Size() -1; i >=0; i--) {
          const auto& kline_obj = data[i];
          bool freezed = HuobiFeedUpdateParser::ParseKline(klines, kline_obj,
              interval, ts);
          to_publish = to_publish || freezed;
        }
      } else {
        NOTREACHED() << "Kline invalid" << ch;
        return;
      }

      if (to_publish) upd_mgr_.PublishKline(native_symbol);
    } else {
      THROW() << "[Huobi Spot] unknown channel: " << channel.type;
    }
  } else if (document.HasMember("rep")) {
    const auto& channel = HuobiChannel::Parse(document["rep"].GetString());
    if (!CheckChannelFilter("market.{}." + channel.type)) {
      if (channel.native_symbol == "wemixusdt" &&
          channel.type == "mbp.150") {
        if (!CheckChannelFilter("market.{}.mbp.5")) {
          upd_mgr_.onEndPacket(packet);
          return;
        }
      } else {
        upd_mgr_.onEndPacket(packet);
        return;
      }
    }

    std::string_view native_symbol = channel.native_symbol;
    if (channel.type.find(R"(mbp.)") == 0 &&
        document.HasMember("data") &&
        (!document["data"].IsNull())) {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        HuobiFeedUpdateParser::ParseSnapshot(book, document, packet.timestamp);
        upd_mgr_.PublishBook();
      }
    }
  } else {
    LOG(INFO) << "[Huobi Spot] meta msg for " << packet.topic_name << ": " << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}  // namespace coin2::exchange::huobi::feed_v1::impl
}  // namespace coin2::exchange::huobi::feed_v1::impl
